package org.ala.tiktools.rocketmq.commons;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 任务线程池
 *
 * @author ala
 * @date 2024-12-03 11:49
 */
public class BizThreadPoolExecutor {

    /**
     *  线程池名称
     */
    protected String name;
    /**
     *  线程池相关信息
     */
    protected int coreSize, maxSize, queueSize;
    /**
     *  当前积压任务数
     */
    protected AtomicInteger backlogNum;
    /**
     *  线程池
     */
    protected ThreadPoolExecutor threadPool;


    /**
     *  初始业务化线程池
     */
    public BizThreadPoolExecutor(String name, int coreSize, int maxSize, int queueSize) {
        this.coreSize = coreSize;
        this.maxSize = maxSize;
        this.queueSize = queueSize;

        //  初始化线程池
        AtomicInteger threadNum = new AtomicInteger(0);
        threadPool = new ThreadPoolExecutor(
                coreSize, maxSize, 180, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(queueSize << 1),
                new ThreadFactory() {
                    public Thread newThread(Runnable r) {
                        Thread t = new Thread(r);
                        t.setDaemon(false);
                        t.setName(name + "_" + threadNum.incrementAndGet());
                        return t;
                    }
                });
        //  初始化积压计数器
        backlogNum = new AtomicInteger();
    }


    /**
     *  吃任务
     */
    public void execute(Runnable run, IBacklog backlog) {
        int bl = backlogNum.incrementAndGet();
        threadPool.submit(() -> {
            try {
                run.run();
            } finally {
                if (backlog != null) {
                    backlog.onComplate(backlogNum.decrementAndGet());
                }
            }
        });
        if (backlog != null) {
            backlog.onSubmit(bl);
        }
    }
    /**
     *  吃任务
     */
    public <R> Future<R> submit(Callable<R> call, IBacklog backlog) {
        int bl = backlogNum.incrementAndGet();
        Future<R> future = threadPool.submit(() -> {
            try {
                return call.call();
            } finally {
                if (backlog != null) {
                    backlog.onComplate(backlogNum.decrementAndGet());
                }
            }
        });
        if (backlog != null) {
            backlog.onSubmit(bl);
        }
        return future;
    }


    /**
     *  积压任务回调
     */
    public static interface IBacklog {
        /**
         *  任务入队列时的回调
         */
        void onSubmit(int backlog);
        /**
         *  任务完成时的回调
         */
        void onComplate(int backlog);
    }


    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getCoreSize() {
        return coreSize;
    }

    public void setCoreSize(int coreSize) {
        this.coreSize = coreSize;
    }

    public int getMaxSize() {
        return maxSize;
    }

    public void setMaxSize(int maxSize) {
        this.maxSize = maxSize;
    }

    public int getQueueSize() {
        return queueSize;
    }

    public void setQueueSize(int queueSize) {
        this.queueSize = queueSize;
    }

    public AtomicInteger getBacklogNum() {
        return backlogNum;
    }

    public void setBacklogNum(AtomicInteger backlogNum) {
        this.backlogNum = backlogNum;
    }

    public ThreadPoolExecutor getThreadPool() {
        return threadPool;
    }

    public void setThreadPool(ThreadPoolExecutor threadPool) {
        this.threadPool = threadPool;
    }
}
